package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2CreateDataFrame {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("df")
      .getOrCreate()


    /**
      * 1、读取csv格式的数据创建df
      */


    val studentDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //列的分割方式
      //指定表结构，必须按照顺序指定
      .schema("id STRING , name STRING, age INT , gender STRING , clazz STRING")
      .load("data/students.txt") //指定读取的路径

    //打印表结构
    studentDF.printSchema()

    //查看数据，默认打印前20行
    studentDF.show()


    /**
      * 2、读取json格式的数据构建DF
      * spark 会自动解析json格式
      *
      */


    val studentJsonDF: DataFrame = spark
      .read
      .format("json")
      .load("data/students.json")

    studentJsonDF.printSchema()
    studentJsonDF.show()

    /**
      * 3、读取jdbc数据
      *
      * 通过网络远程读取mysql中的数据
      */
    val jdbcDF: DataFrame = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable", "bigdata.students")
      .option("user", "root")
      .option("password", "123456")
      .load()

    jdbcDF.printSchema()
    jdbcDF.show()


    /**
      * 4、读取parquet格式的数据
      * parquet格式的数据中自带列名和列的类型，
      * parquet会对数据进行压缩，体积变小，解压和压缩需要时间
      *
      */


    //保存一个parquet格式的文件
    studentDF
      .write
      .format("parquet")
      .mode(SaveMode.Overwrite)
      .save("data/parquet")

    //读取parquet格式的数据
    val parquetDF: DataFrame = spark
      .read
      .format("parquet")
      .load("data/parquet")

    parquetDF.printSchema()
    parquetDF.show()

  }

}
